#YashanDB Flink CDC Connector

YashanDB CDC连接器允许从YashanDB数据库读取快照数据和增量数据。

本文档将介绍如何配置YashanDB数据库和YashanDB CDC连接器实现对YashanDB数据库进行捕获变更事件。

# 版本配套说明

# 前提条件

• 已安装Flink，具体操作请查阅Flink官方文档 (opens new window)。

• 已获取下列jar包并存放至<FLINK_HOME>/lib/目录下：

  可自行使用构建自动化工具（例如Maven）将项目构建打包，也可联系我们的技术支持获取。

  • flink-connector-yashandb.jar

  • YStream-vxx.xx.xx.jar

  • yashandb-jdbc.jar

# 使用限制

• 数据类型不能为JSON，XMLTYPE，自定义数据类型，ST_GEOMETRY或BOX2D。
• 不能为带LOB数据类型的无主键表（受Ystream相关约束，无主键表的update dml和delete dml的LOB值会丢失）。
• 不支持增量DDL。
• 如果使用flink stream API，每个任务的表数量不得超过一万。
• Ystream服务最多32个，因此flink的任务最多同时只能运行32个。

# 配置YashanDB

为了让连接器能够正常运行以及监视表的相关信息变化，您需要开启YashanDB的附加日志，并为连接器的连接用户授予相应权限。

# 步骤1：配置Ystream内存池

增量数据依赖YashanDB的YStream实时获取YashanDB的已提交数据。

当您使用YashanDB作为源使用含增量同步的任务时，您需要在YashanDB中为YStream分配内存池：

ALTER SYSTEM SET STREAM_POOL_SIZE = streamPoolSize;

# 步骤2：开启附加日志

读取增量数据变更需要开启附加日志，可按需开启库级或表级附加日志。

Caution：

不开启附加日志或开启附加日志的对象不正确会导致数据丢失甚至任务失败。

当您需要监听库下全部对象时（包含新增对象），可开全库附加日志，方式如下：

ALTER DATABASE ADD SUPPLEMENTAL LOG TABLE TYPE (HEAP);
ALTER DATABASE ADD SUPPLEMENTAL LOG DATA (PRIMARY KEY) COLUMNS;

当您仅需要监听某些表时，可开启表级附加日志，方式如下：

ALTER TABLE tablename ADD SUPPLEMENTAL LOG DATA (PRIMARY KEY) COLUMNS;

# 步骤3：授予相关用户权限

YashanDB CDC connector的连接用户需要相关权限，请授予如下权限保障任务能够正常运行：

GRANT CREATE SESSION TO username;
GRANT SELECT ON V_$DATABASE TO username;
GRANT SELECT ON V_$TRANSACTION TO username;
GRANT SELECT ON V_$YSTREAM_SERVER TO username;
GRANT FLASHBACK ANY TABLE TO username; 
GRANT SELECT ANY TABLE TO username;
GRANT YSTREAM_CAPTURE TO username;
GRANT SELECT ON DBA_YSTREAM_PARAMETERS TO username;

# 创建YashanDB CDC表

YashanDB CDC表可以定义如下：

-- register an YashanDB table 'products' in Flink SQL
Flink SQL> CREATE TABLE products (
     ID INT NOT NULL,
     NAME STRING,
     DESCRIPTION STRING,
     WEIGHT DECIMAL(10, 3),
     PRIMARY KEY(id) NOT ENFORCED
     ) WITH (
     'connector' = 'yashandb-cdc',
     'hostname' = 'localhost',
     'port' = '1688',
     'username' = 'flinkuser',
     'password' = 'flinkpw',
     'ystream.serverName'='server1',
     'schema-name' = 'inventory',
     'table-name' = 'products');
  
-- read snapshot and redo logs from products table
Flink SQL> SELECT * FROM products;

# 连接器参数

Note:

关于YStream的使用介绍可查阅：

• YStream参考手册

• DBMS_YSTREAM_ADM高级包

# 功能特性

# Exactly-Once Processing

YashanDB CDC connector首先读取快照阶段，然后再精确一致性读取变更数据事件，即使中途中发生任务失败，依赖flink的checkpoint或savepoint也会从失败点位或指定点位进行重新拉取增量任务。

# 启动读取位置

通过配置选项【scan.startup.mode】可设置连接器的启动模式：

• initial（default）: 先启动全量快照读取，再启动增量读取redo log捕获更改事件。

• latest-offset：不启动快照读取，直接从当前日志点位增量读取redo log捕获更改事件。

# 单线程增量读取

YashanDB CDC增量源无法并行读取，因为只能一个任务可以接收更改事件。

# DataStream Source

YashanDB CDC连接器也可以是DataStream源，示例如下：

import com.sics.flink.connector.yashandb.config.StartupOptions;
import com.sics.flink.connector.yashandb.internal.options.YstreamOptions;
import com.sics.flink.connector.yashandb.source.YashanDBIncrementalSource;
import com.sics.flink.connector.yashandb.source.cdc.deserialization.JsonYstreamDeserializationSchema;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.ReadableConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class TestYashanDBCdc {
  public static void main(String[] args) throws Exception {
    ReadableConfig config = new Configuration();

    YashanDBIncrementalSource<String> source =
        YashanDBIncrementalSource.<String>newBuilder()
            .hostname("hostname")
            .port(1688)
            .fetchSize(1024)
            .schemaList("TEST")
            .startupOptions(StartupOptions.initial())
            .deserializer(new JsonYstreamDeserializationSchema())
            .username("username")
            .password("password")
            .tableList("TEST1.TAB01")
            .ystreamOptions(YstreamOptions.defaultOption) // ystream option
            .ystreamServerName("server1")
            .build();
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

    env.fromSource(source, WatermarkStrategy.noWatermarks(), "yashandb source")
        .setParallelism(4)
        .print()
        .setParallelism(1);
    env.execute("Print yashadnb Snapshot + RedoLog");
  }
}

# 数据类型映射

# 常见问题

# Q1. 为什么运行yashandb-cdc需要YStream依赖？

yashandb-cdc内部使用了YashanDB的YStream来捕获增量数据信息，yashandb-cdc依赖包不包含YStream依赖，需要用户自行手动拉取YStream依赖放入flink的lib目录下。

# Q2. yashandb-cdc报错“YashanDB YStream serverName 'server' has existed in database, Please enter a non-existent option 'ystream.serverName' in V_$YSTREAM_SERVER.”该怎么处理？

• 首次启动yashandb-cdc，会根据用户填写的选项ystream.serverName自动创建YashanDB数据库的Ystream服务，如果数据库中已存在同名YStream服务会创建失败并报此错，需修改ystream.serverName值并确保全局唯一。

• 非首次启动yashandb-cdc：

  • 如果使用savepoint启动，yashandb-cdc会复用上次创建的YStream服务重新从上一个数据点位拉取任务，不涉及新建YStream服务，不会出现此报错。

  • 若用户重新开启新的cdc任务，需重新配置新的唯一ystream.serverName值。

为降低重名报错的复现率，可在确认YStream服务已无需再使用后自行删除，删除语句参考如下：

EXEC DBMS_YSTREAM_ADM.STOP('ystream_server');

EXEC DBMS_YSTREAM_ADM.DROP('ystream_server');

# Q3. 运行yashandb-cdc报错“The number of YStream server in the database has reached 32, and YashanDB's YStream server supports a maximum of 32. Please manually execute 'EXEC DBMS_YSTREAM_ADM.DROP('ystream_server')' to delete the YStream server in' select * from V_$YSTREAM_SERVER'”该如何处理？

该错误信息表示YashanDB数据库中的YStream服务数量已达最大值32个，可查看V_$YSTREAM_SERVER视图获取所有YStream服务信息，并结合实际需求手动清理未使用/无需再用的YStream服务。